otap: unify real OtapPdata processor, live reconfiguration, and YAML pipeline E2E - #9

…eal OTAP workspace

Per request: actually get an OTAP Dataflow checkout and compile/test
against it, rather than leaving otap_bridge.rs's binding as "read from
source, never compiled." Staged both files into a real clone of
open-telemetry/otel-arrow @ 3e85c3460361446ebfce99e9f35fffd2dd5ab740
(2026-08-24) as a `crates/asap-sketches-registry` workspace member
(path-depping back to asap-precompute-rs exactly as Cargo.toml's own
doc describes) and ran the real cargo build/clippy/fmt/test there.

## Bugs the real compiler caught (none of these were guessable from
## source reading alone)

- `ProcessorFactory::create`'s function-pointer type gained a fifth
  parameter, `capabilities: &capability::registry::Capabilities`
  ("per-node, one-shot view of extension capabilities... factories
  that don't depend on any extension can ignore the parameter") —
  not present in whatever version this adapter was originally read
  against. Added as an unused parameter to
  create_asap_sketches_processor.
- Two borrow-checker errors in the OtapPdata decode path: `if let
  Some(gauge) = data.as_gauge() { gauge.data_points().collect() }`
  doesn't compile because the collected Vec's items borrow from
  `gauge`, which is dropped at the end of the `if let` arm. Fixed by
  factoring the per-data-point body into a `DecodeAccumulator` struct
  with a generic `push_data_point<D: NumberDataPointView>` method,
  called inline from each of the Gauge/Sum branches instead of trying
  to unify their different concrete NumberDataPointView types into
  one Vec first.
- One unused import (`TryFromWithOptions`) clippy caught immediately.

## What's now genuinely verified (not just "compiles")

Added 3 new tests to otap_bridge.rs exercising the real
OtapArrowRecords::Metrics / OtapPdata / encode_metrics_otap_batch /
OtapMetricsView machinery end to end:
- encode_then_decode_round_trips_a_scalar_metric
- encode_then_decode_round_trips_a_sketch_envelope_carried_as_a_metric_attribute
  (the "self-describing sketch binary inside an OTAP metric" case --
  an `_asap_envelope` Bytes attribute survives the round trip
  byte-for-byte)
- decode_returns_none_records_for_zero_rows

All 10 tests in the staged crate pass (7 pre-existing config-shape
tests + 3 new), `cargo clippy -D warnings` and `cargo fmt --check`
both clean, against the real workspace.

## Not changed

Applied `cargo fmt`'s reformatting (this workspace's rustfmt config
differs slightly from what the file was originally written against --
import grouping mainly) back onto the canonical otap-patch/ copies.
Updated mod.rs's, otap_bridge.rs's, and README.md's doc comments from
"unverified" to reflect what's actually confirmed now, and precisely
what "verified" means here: staged into a *separate*, temporary
checkout of the real workspace, not something this repo's own build
wires up -- otap-patch/ itself still has no standalone build in this
repo.

Added .gitignore entry for /otel-arrow/ (the temporary checkout used
for this verification, not committed).

…receiver nodes

Both otap_wire.rs's transport (OtapWireWriter/OtapWireReader) and the
wire lane's receive side had no consumer inside the crate — this closes
that: real OTAP nodes now actually drive them at runtime, not just tests.

- New otap_receiver.rs: AsapSketchesReceiver, a real
  local::Receiver<OtapPdata> (not local::Processor — that trait is purely
  reactive and can't independently accept a TCP connection). Listens on a
  configured address, decodes each connection with OtapWireReader, pushes
  into the pipeline via effect_handler.send_message. Registered under
  urn:asap:receiver:asap_sketches_wire via OTAP_RECEIVER_FACTORIES.
- AsapSketchesProcessor gains an optional peer_addr config field:
  emit_envelopes now forwards over a lazily-connected, persistent
  OtapWireWriter connection to that peer when set, falling back to the
  existing generic pipeline hop otherwise. A send failure drops the
  window and resets the connection so the next window reconnects.
- otap_wire.rs's #![allow(dead_code)] is gone — both OtapWireWriter and
  OtapWireReader now have real, non-test callers.
- New end-to-end test: the real AsapSketchesReceiver::start() bound to a
  real loopback TCP socket, fed by a real OtapWireWriter::send from a
  connected client, decoded and pushed through the actual pipeline
  machinery (OTAP's own TestRuntime harness) — not a mock.

Verified against the same staged open-telemetry/otel-arrow checkout
(3e85c3460361446ebfce99e9f35fffd2dd5ab740, 2026-08-24): build, clippy
-D warnings, fmt --check, and test all clean; 16/16 tests passing
(up from 12/12), including the new end-to-end receiver test.
asap-precompute-rs's own suite: 156/156, unaffected.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

… already gives us dictionary economics

The direct-TCP wire lane added in the previous commit (otap_wire.rs +
otap_receiver.rs's AsapSketchesReceiver, peer_addr config) is gone.
There is now exactly one transport: effect_handler.send_message_with_source_node,
i.e. whatever pipeline this node is already wired into.

Why: the wire lane existed to give sketch traffic real dictionary/
schema-reuse economics (asap_precompute_rs::otap::dictionary's SCHEMA/
DICTIONARY/RECORD tiering, from PR #5/#6) that riding a generic OTLP
metric didn't have. But that's solving a problem OTAP's real Arrow
encoding already solves — otap_bridge's encode_metrics_otap_batch
dictionary-encodes the metric name and every string-valued attribute
key/value by construction. Verified against the real staged workspace:

  === payload_type UnivariateMetrics ===
    name : Dictionary(UInt8, Utf8)
  === payload_type NumberDpAttrs ===
    parent_id : Dictionary(UInt8, UInt32)
    key       : Dictionary(UInt8, Utf8)
    str       : Dictionary(UInt16, Utf8)

That's the same "send the dictionary once, reference it after that"
shape SeriesDictionary was reinventing at the application layer, done
instead at the columnar/Arrow-IPC level — with no dictionary state for
this adapter to track, no second wire protocol, and none of the
"must be one ordered, single-consumer stream or the series_id
reference dangles" correctness constraint a hand-rolled scheme has.

Added a permanent regression test guarding this fact
(otap_bridge::tests::real_otap_encoding_dictionary_encodes_metric_name_and_string_attributes)
so a silent upstream schema change surfaces loudly rather than being
rediscovered from scratch. asap_precompute_rs::otap::dictionary
(SeriesDictionary / SketchStreamBatch) stays in the tree, tested, and
still backs the legacy asap_precompute_rs::otap::wire example
binaries — it's just not part of this adapter's path.

Re-verified against the same staged open-telemetry/otel-arrow checkout
(3e85c3460361446ebfce99e9f35fffd2dd5ab740, 2026-08-24): build, clippy
-D warnings, fmt --check, and test all clean; 11/11 tests passing.
asap-precompute-rs's own suite: 156/156, unaffected.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

…e, real OtapPdata everywhere

Full merge, per explicit direction: no more otap-patch/ overlay
requiring manual staging into a checked-out OTAP Dataflow workspace to
compile. asap-precompute-rs is now the ONE crate, and depends directly
on the real otel-arrow-dfe-* crates (plain git dependency, pinned to
3e85c3460361446ebfce99e9f35fffd2dd5ab740, same pattern already used
for asap_sketchlib) behind a new `otap-engine` Cargo feature.
`cargo build --features otap-engine` just works — no staging script,
no temporary checkout, no copying files by hand.

New modules (moved + rewritten from otap-patch/all/{mod.rs,otap_bridge.rs}):

- otap::processor — AsapSketchesProcessor, the real
  local::Processor<OtapPdata> node, linkme-registered under
  urn:asap:processor:asap_sketches.
- otap::codec (renamed from the working name `bridge` — it's not
  bridging two representations anymore, see below) — the real
  SketchEnvelope <-> OtapPdata binding.

The bigger change is what codec.rs does differently from the old
otap_bridge.rs it replaces: it builds/reads a real OtapPdata *directly*
from/to &[SketchEnvelope], skipping the intermediate flat RecordBatch
(encode_batch) and OtapMetricRecords two-batch family (lift) hops
entirely. Those three representations for one job only existed because
of speculative multi-adapter generality (Telegraf/Vector) that never
materialized in this repo, which only ever ships to OTAP -- so codec.rs
implements OTAP's own MetricsView trait family straight over envelope
slices instead. encode_batch/decode_batch/records::{flatten,lift}
still exist, still tested, and still back the legacy
SeriesDictionary/otap::wire transport and its standalone example
binaries -- they're just not part of the otap-engine path anymore.

A real bug found and fixed along the way: encode_batch unconditionally
sets `_asap_envelope` even for estimate-mode envelopes (empty
payload), and arrow_array::BinaryArray treats `Some(&[])` as present-
not-null -- confirmed empirically -- so an estimate-mode gauge would
misroute through the envelope decode path with an empty payload
instead of the scalar path. codec.rs's direct encoder only attaches
_asap_envelope (and its sibling attributes) when the payload is
actually non-empty, with a defensive filter on decode too. New test:
estimate_mode_envelope_round_trips_as_a_scalar_not_an_empty_envelope.
The legacy encode_batch/decode_batch pair keeps the original behavior
untouched -- fixing it there was out of scope for this rewrite.

Also moved plugins/asap_sketches/{README.md,sample.toml} to
asap-precompute-rs/plugins/asap_sketches/ (content updated to match:
no more Phase D/"deliberately deferred" language, since it's done),
dropped the empty src/mod.rs placeholder it predated, and rewrote the
crate-root and repo-root README/module docs throughout to describe the
one-crate reality instead of the old Layer A/otap-patch split.

Verified via this repo's own `cargo build/test/clippy/fmt` at all
three feature levels (default, otap, otap-engine) -- no staging into
any external checkout: 100/113/169 tests pass respectively (default/
otap/otap-engine), clippy -D warnings clean, fmt clean.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>